//! # 集成测试和端到端测试
//!
//! 提供消息总线系统的完整集成测试和端到端场景测试

#[cfg(test)]
mod integration_tests {
    use crate::prelude::*;
    use crate::mock::{MockMessageBus, test_helpers::*};
    use crate::publisher::SimplePublisher;
    use serde_json::json;
    use std::sync::Arc;
    use tokio::time::Duration;

    /// 完整流程集成测试
    mod end_to_end_tests {
        use super::*;

        #[tokio::test]
        async fn test_complete_message_flow() {
            let bus = Arc::new(MockMessageBus::new());
            
            // 1. 创建发布者
            let publisher = SimplePublisher::new(bus.clone(), "integration_publisher".to_string());
            
            // 2. 创建订阅者
            let subscriber = bus.subscribe("integration.test.events").await.unwrap();
            
            // 3. 发布消息
            let message = Message::new(
                "integration.test.events",
                json!({
                    "event_type": "user_registration",
                    "user_id": "user_123",
                    "timestamp": chrono::Utc::now().to_rfc3339(),
                    "metadata": {
                        "source": "web_app",
                        "version": "1.0.0"
                    }
                })
            );
            
            // 设置correlation_id
            let mut message = message;
            message.metadata.correlation_id = Some(uuid::Uuid::new_v4());
            
            publisher.publish(message.clone()).await.unwrap();
            
            // 4. 接收和处理消息
            let received = subscriber.receive_timeout(1000).await.unwrap();
            assert!(received.is_some());
            
            let received_msg = received.unwrap();
            assert_eq!(received_msg.topic(), "integration.test.events");
            assert_eq!(received_msg.payload()["event_type"], "user_registration");
            assert_eq!(received_msg.payload()["user_id"], "user_123");
            
            // 5. 确认消息
            subscriber.ack(&received_msg).await.unwrap();
            
            // 6. 验证统计信息
            let stats = bus.stats().await.unwrap();
            assert_eq!(stats.total_published, 1);
            assert_eq!(stats.active_subscribers, 1);
            
            // 7. 清理资源
            bus.unsubscribe(subscriber.subscription_id()).await.unwrap();
            publisher.close().await.unwrap();
            
            let final_stats = bus.stats().await.unwrap();
            assert_eq!(final_stats.active_subscribers, 0);
        }

        #[tokio::test]
        async fn test_multi_subscriber_workflow() {
            let bus = Arc::new(MockMessageBus::new());
            
            // 创建多个订阅者处理不同的消息类型
            let user_subscriber = bus.subscribe("events.user.*").await.unwrap();
            let order_subscriber = bus.subscribe("events.order.*").await.unwrap();
            let audit_subscriber = bus.subscribe("events.**").await.unwrap(); // 审计订阅所有事件
            
            let publisher = SimplePublisher::new(bus.clone(), "multi_pub".to_string());
            
            // 发布不同类型的事件
            let events = vec![
                ("events.user.created", json!({"user_id": "u1", "action": "create"})),
                ("events.user.updated", json!({"user_id": "u1", "action": "update"})),
                ("events.order.placed", json!({"order_id": "o1", "amount": 100})),
                ("events.order.cancelled", json!({"order_id": "o1", "reason": "timeout"})),
            ];
            
            for (topic, payload) in events {
                let message = Message::new(topic, payload);
                publisher.publish(message).await.unwrap();
            }
            
            // 验证消息分发
            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), 4);
            
            // 验证订阅者数量
            assert_eq!(bus.subscriber_count("events.user.created").await.unwrap(), 2); // user + audit
            assert_eq!(bus.subscriber_count("events.order.placed").await.unwrap(), 2); // order + audit
        }

        #[tokio::test] 
        async fn test_error_handling_integration() {
            let bus = Arc::new(MockMessageBus::new());
            let publisher = SimplePublisher::new(bus.clone(), "error_pub".to_string());
            let subscriber = bus.subscribe("error.test").await.unwrap();
            
            // 正常消息
            let normal_msg = Message::new("error.test", json!({"type": "normal"}));
            assert!(publisher.publish(normal_msg).await.is_ok());
            
            // 模拟发布错误
            bus.set_behavior("publish", crate::mock::MockBehavior::new().with_failure("发布失败")).await;
            let error_msg = Message::new("error.test", json!({"type": "error"}));
            assert!(publisher.publish(error_msg).await.is_err());
            
            // 恢复正常
            bus.set_behavior("publish", crate::mock::MockBehavior::new()).await;
            let recovery_msg = Message::new("error.test", json!({"type": "recovery"}));
            assert!(publisher.publish(recovery_msg).await.is_ok());
            
            // 验证错误恢复后系统正常工作
            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), 2); // normal + recovery
        }
    }

    /// 组件集成测试
    mod component_integration_tests {
        use super::*;

        #[tokio::test]
        async fn test_publisher_subscriber_integration() {
            let bus = Arc::new(MockMessageBus::new());
            
            // 测试不同类型的发布者和订阅者组合
            let simple_pub = SimplePublisher::new(bus.clone(), "simple".to_string());
            let batch_pub = crate::publisher::BatchPublisher::new(
                bus.clone(),
                "batch".to_string(),
                5,
                1000
            );
            
            let wildcard_sub = bus.subscribe("test.*").await.unwrap();
            let exact_sub = bus.subscribe("test.exact").await.unwrap();
            
            // Simple Publisher 发布
            let simple_msg = Message::new("test.simple", json!({"from": "simple"}));
            simple_pub.publish(simple_msg).await.unwrap();
            
            // Batch Publisher 发布
            let batch_msg = Message::new("test.exact", json!({"from": "batch"}));
            batch_pub.publish(batch_msg).await.unwrap();
            batch_pub.flush().await.unwrap();
            
            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), 2);
        }

        #[tokio::test]
        async fn test_event_system_integration() {
            let bus = Arc::new(MockMessageBus::new());
            
            // 创建事件处理链
            let event_subscriber = bus.subscribe("events.user.**").await.unwrap();
            let audit_subscriber = bus.subscribe("audit.**").await.unwrap();
            
            let publisher = SimplePublisher::new(bus.clone(), "event_pub".to_string());
            
            // 发布用户事件
            let user_event = Message::event(
                "events.user.registration.completed",
                json!({
                    "user_id": "12345",
                    "email": "user@example.com",
                    "timestamp": chrono::Utc::now().to_rfc3339()
                })
            );
            
            publisher.publish(user_event).await.unwrap();
            
            // 发布审计事件
            let audit_event = Message::event(
                "audit.user.action.login",
                json!({
                    "user_id": "12345",
                    "action": "login",
                    "ip": "192.168.1.100"
                })
            );
            
            publisher.publish(audit_event).await.unwrap();
            
            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), 2);
            
            // 验证事件类型
            for msg in published {
                assert_eq!(msg.metadata.attributes.message_type, MessageType::Event);
            }
        }

        #[tokio::test]
        async fn test_persistence_integration() {
            let bus = Arc::new(MockMessageBus::new());
            let subscriber = bus.subscribe("persistent.messages").await.unwrap();
            let publisher = SimplePublisher::new(bus.clone(), "persist_pub".to_string());
            
            // 发布需要持久化的消息
            let mut persistent_msg = Message::new(
                "persistent.messages",
                json!({"important": "data"})
            );
            persistent_msg.metadata.attributes.persistence = PersistenceLevel::Disk;
            
            publisher.publish(persistent_msg).await.unwrap();
            
            // 发布临时消息
            let mut transient_msg = Message::new(
                "persistent.messages",
                json!({"temp": "data"})
            );
            transient_msg.metadata.attributes.persistence = PersistenceLevel::None;
            
            publisher.publish(transient_msg).await.unwrap();
            
            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), 2);
            assert_eq!(published[0].metadata.attributes.persistence, PersistenceLevel::Disk);
            assert_eq!(published[1].metadata.attributes.persistence, PersistenceLevel::None);
        }
    }

    /// 实际场景端到端测试
    mod real_scenario_e2e_tests {
        use super::*;

        #[tokio::test]
        async fn test_order_processing_workflow() {
            let bus = Arc::new(MockMessageBus::new());
            
            // 创建订单处理工作流的订阅者
            let order_service = bus.subscribe("orders.*").await.unwrap();
            let payment_service = bus.subscribe("payments.*").await.unwrap();
            let inventory_service = bus.subscribe("inventory.*").await.unwrap();
            let notification_service = bus.subscribe("notifications.*").await.unwrap();
            
            let publisher = SimplePublisher::new(bus.clone(), "order_system".to_string());
            
            // 模拟订单处理流程
            let order_id = uuid::Uuid::new_v4();
            let correlation_id = uuid::Uuid::new_v4();
            
            // 1. 订单创建
            let mut order_created = Message::new(
                "orders.created",
                json!({
                    "order_id": order_id,
                    "user_id": "user_123",
                    "items": [{"sku": "ITEM001", "quantity": 2}],
                    "total": 49.99
                })
            );
            order_created.metadata.correlation_id = Some(correlation_id);
            
            publisher.publish(order_created).await.unwrap();
            
            // 2. 库存检查
            let mut inventory_check = Message::new(
                "inventory.check",
                json!({
                    "order_id": order_id,
                    "items": [{"sku": "ITEM001", "quantity": 2}]
                })
            );
            inventory_check.metadata.correlation_id = Some(correlation_id);
            
            publisher.publish(inventory_check).await.unwrap();
            
            // 3. 支付处理
            let mut payment_request = Message::new(
                "payments.process",
                json!({
                    "order_id": order_id,
                    "amount": 49.99,
                    "payment_method": "credit_card"
                })
            );
            payment_request.metadata.correlation_id = Some(correlation_id);
            
            publisher.publish(payment_request).await.unwrap();
            
            // 4. 通知发送
            let mut notification = Message::new(
                "notifications.send",
                json!({
                    "user_id": "user_123",
                    "type": "order_confirmation",
                    "order_id": order_id
                })
            );
            notification.metadata.correlation_id = Some(correlation_id);
            
            publisher.publish(notification).await.unwrap();
            
            // 验证工作流
            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), 4);
            
            // 验证所有消息都有相同的correlation_id
            for msg in published {
                assert_eq!(msg.metadata.correlation_id, Some(correlation_id));
            }
        }

        #[tokio::test]
        async fn test_user_activity_monitoring() {
            let bus = Arc::new(MockMessageBus::new());
            
            // 创建用户活动监控系统
            let activity_logger = bus.subscribe("user.activity.**").await.unwrap();
            let security_monitor = bus.subscribe("user.security.*").await.unwrap();
            let analytics_service = bus.subscribe("analytics.**").await.unwrap();
            
            let publisher = SimplePublisher::new(bus.clone(), "user_system".to_string());
            
            let user_id = "user_456";
            let session_id = uuid::Uuid::new_v4();
            
            // 用户登录
            let login_event = Message::event(
                "user.activity.login",
                json!({
                    "user_id": user_id,
                    "session_id": session_id,
                    "ip": "192.168.1.100",
                    "user_agent": "Mozilla/5.0...",
                    "timestamp": chrono::Utc::now().to_rfc3339()
                })
            );
            publisher.publish(login_event).await.unwrap();
            
            // 页面访问
            let page_view = Message::event(
                "user.activity.page_view",
                json!({
                    "user_id": user_id,
                    "session_id": session_id,
                    "page": "/dashboard",
                    "referrer": "/login"
                })
            );
            publisher.publish(page_view).await.unwrap();
            
            // 安全事件（多次失败的密码重置）
            let security_event = Message::event(
                "user.security.suspicious_activity",
                json!({
                    "user_id": user_id,
                    "event": "multiple_password_reset_attempts",
                    "count": 5,
                    "time_window": "5_minutes"
                })
            );
            publisher.publish(security_event).await.unwrap();
            
            // 分析事件
            let analytics_event = Message::event(
                "analytics.user.engagement",
                json!({
                    "user_id": user_id,
                    "session_duration": 1800,
                    "pages_viewed": 5,
                    "actions_performed": 12
                })
            );
            publisher.publish(analytics_event).await.unwrap();
            
            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), 4);
        }

        #[tokio::test]
        async fn test_iot_data_processing_pipeline() {
            let bus = Arc::new(MockMessageBus::new());
            
            // IoT数据处理管道
            let data_ingestion = bus.subscribe("sensors.raw.**").await.unwrap();
            let data_processing = bus.subscribe("processing.*").await.unwrap();
            let alert_system = bus.subscribe("alerts.**").await.unwrap();
            let dashboard = bus.subscribe("dashboard.updates.*").await.unwrap();
            
            let publisher = SimplePublisher::new(bus.clone(), "iot_gateway".to_string());
            
            let device_id = "sensor_001";
            let timestamp = chrono::Utc::now();
            
            // 原始传感器数据
            let raw_data = Message::new(
                "sensors.raw.temperature",
                json!({
                    "device_id": device_id,
                    "temperature": 75.5,
                    "humidity": 60.2,
                    "timestamp": timestamp.to_rfc3339(),
                    "location": "warehouse_a"
                })
            );
            publisher.publish(raw_data).await.unwrap();
            
            // 处理后的数据
            let processed_data = Message::new(
                "processing.aggregated",
                json!({
                    "device_id": device_id,
                    "avg_temperature": 74.8,
                    "trend": "stable",
                    "period": "1_hour"
                })
            );
            publisher.publish(processed_data).await.unwrap();
            
            // 温度警报
            let temperature_alert = Message::new(
                "alerts.temperature.high",
                json!({
                    "device_id": device_id,
                    "current_temp": 85.0,
                    "threshold": 80.0,
                    "severity": "warning"
                })
            );
            publisher.publish(temperature_alert).await.unwrap();
            
            // 仪表板更新
            let dashboard_update = Message::new(
                "dashboard.updates.realtime",
                json!({
                    "device_id": device_id,
                    "metrics": {
                        "temperature": 75.5,
                        "humidity": 60.2,
                        "status": "normal"
                    }
                })
            );
            publisher.publish(dashboard_update).await.unwrap();
            
            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), 4);
        }
    }

    /// 系统级集成测试
    mod system_integration_tests {
        use super::*;
        use std::sync::atomic::{AtomicUsize, Ordering};

        #[tokio::test]
        async fn test_full_system_lifecycle() {
            let bus = Arc::new(MockMessageBus::new());
            
            // 系统启动阶段
            let startup_subscriber = bus.subscribe("system.startup.**").await.unwrap();
            let health_subscriber = bus.subscribe("system.health.*").await.unwrap();
            
            let publisher = SimplePublisher::new(bus.clone(), "system".to_string());
            
            // 发布启动事件
            let startup_events = vec![
                "system.startup.database.connected",
                "system.startup.cache.initialized", 
                "system.startup.services.loaded",
                "system.startup.ready"
            ];
            
            for event in startup_events {
                let message = Message::event(event, json!({"timestamp": chrono::Utc::now().to_rfc3339()}));
                publisher.publish(message).await.unwrap();
            }
            
            // 运行阶段 - 健康检查
            let health_check = Message::new(
                "system.health.check",
                json!({
                    "services": {
                        "database": "healthy",
                        "cache": "healthy", 
                        "api": "healthy"
                    },
                    "memory_usage": "75%",
                    "cpu_usage": "45%"
                })
            );
            publisher.publish(health_check).await.unwrap();
            
            // 关闭阶段
            let shutdown_subscriber = bus.subscribe("system.shutdown.**").await.unwrap();
            
            let shutdown_events = vec![
                "system.shutdown.services.stopping",
                "system.shutdown.connections.closing",
                "system.shutdown.complete"
            ];
            
            for event in shutdown_events {
                let message = Message::event(event, json!({"timestamp": chrono::Utc::now().to_rfc3339()}));
                publisher.publish(message).await.unwrap();
            }
            
            // 验证完整的生命周期
            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), 8); // 4 startup + 1 health + 3 shutdown
            
            // 清理
            bus.close().await.unwrap();
        }

        #[tokio::test]
        async fn test_high_load_system_integration() {
            let bus = Arc::new(MockMessageBus::new());
            let message_count = Arc::new(AtomicUsize::new(0));
            
            // 创建多个服务模拟高负载
            let services = vec!["auth", "user", "order", "payment", "notification"];
            let mut subscribers = vec![];
            
            for service in &services {
                let pattern = format!("{}.**", service);
                let sub = bus.subscribe(&pattern).await.unwrap();
                subscribers.push(sub);
            }
            
            let publisher = SimplePublisher::new(bus.clone(), "load_test".to_string());
            let publisher = Arc::new(publisher);
            let mut handles = vec![];
            
            // 每个服务并发发送大量消息
            for service in &services {
                let pub_clone = publisher.clone();
                let counter = message_count.clone();
                let service_name = service.clone();
                
                let handle = tokio::spawn(async move {
                    for i in 0..200 {
                        let topic = format!("{}.events.action_{}", service_name, i);
                        let message = Message::new(&topic, json!({
                            "service": service_name,
                            "action_id": i,
                            "timestamp": chrono::Utc::now().to_rfc3339()
                        }));
                        
                        if pub_clone.publish(message).await.is_ok() {
                            counter.fetch_add(1, Ordering::Relaxed);
                        }
                    }
                });
                handles.push(handle);
            }
            
            // 等待所有任务完成
            for handle in handles {
                handle.await.unwrap();
            }
            
            let total_sent = message_count.load(Ordering::Relaxed);
            assert_eq!(total_sent, 1000); // 5 services * 200 messages
            
            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), 1000);
            
            let stats = bus.stats().await.unwrap();
            assert_eq!(stats.total_published, 1000);
            assert_eq!(stats.active_subscribers, 5);
        }
    }
}